#include <ddm/MapReducer.h>
#include <ddm/Util.h>
#include <folly/IPAddress.h>
#include <folly/executors/GlobalExecutor.h>
#include <folly/logging/xlog.h>
#include <openr/common/NetworkUtil.h>
#include <openr/common/OpenrClient.h>
#include <openr/if/gen-cpp2/OpenrCtrlCppAsyncClient.h>
#include <openr/kvstore/KvStoreClientInternal.h>
#include <openr/kvstore/KvStoreUtil.h>

#include <iostream>

namespace ddm {
MapReducer::MapReducer(
    openr::messaging::RQueue<AtomicRoute> atomicRouteUpdateQueue,
    openr::messaging::RQueue<ddm::thrift::Requirement> requirementsQueue,
    std::string nodeId)
    : nodeId_(nodeId) {

  // Fiber to process atomic route updates
  addFiberTask(
      [q = std::move(atomicRouteUpdateQueue), this]() mutable noexcept {
        while (true) {
          auto maybeThriftObj = q.get();  // perform read
          if (maybeThriftObj.hasError()) {
            XLOG(DBG1) << "Terminating atomic route processing fiber";
            break;
          }

          processAtomicRouteUpdate(std::move(maybeThriftObj).value());
        }
      });
  addFiberTask([q = std::move(requirementsQueue), this]() mutable noexcept {
    while (true) {
      auto maybeThriftObj = q.get();  // perform read
      if (maybeThriftObj.hasError()) {
        XLOG(DBG1) << "Terminating requirements processing fiber";
        break;
      }

      processRequirement(std::move(maybeThriftObj).value());
    }
  });
}

void MapReducer::processAtomicRouteUpdate(AtomicRoute&& atomicRoute) {
  std::cout << atomicRoute.egress() << ": ";
  for (auto& range : atomicRoute.ranges()) {
    std::cout << range.first << "," << range.second << ",";
  }
  std::cout << std::endl;
  // egressToAtomicRoute_[atomicRoute.egress()] = std::move(atomicRoute);
  egressToRanges_[atomicRoute.egress()] = atomicRoute.ranges();
}

void MapReducer::processRequirement(ddm::thrift::Requirement&& req) {
  std::cout << "Got requirement MR: " << *req.name()
            << " val: " << openr::toIPNetwork(*req.prefix()).first.str() << " "
            << *req.prefix()->prefixLength() << std::endl;
  folly::CIDRNetwork hs = openr::toIPNetwork(*req.prefix());
  auto range = ddm::computeRange(hs);
  std::cout << range.first << "," << range.second << std::endl;

  reqToAtomicRoutes_[*req.name()] = map(hs);

  // for (auto& [r, ar] : reqToAtomicRoutes_) {
  //   for (auto& [egress, hs] : ar) {
  //     kvStoreClient_->setKey(openr::AreaId{"0"},
  //                            "hs:" + nodeId_ + ":" + r + ":" + egress, "111");
  //   }
  // }

  reduce({range});
}

void MapReducer::map1(ddm::thrift::Requirement& req) {
  if (*req.name() == "reqnhop") {
    folly::CIDRNetwork hs = openr::toIPNetwork(*req.prefix());
    auto range = ddm::computeRange(hs);
    reqToTuple_[*req.name()] = std::make_pair<RANGES, int>({range}, 0);
  }
}

void MapReducer::transition(std::string reqName) {
  std::map<std::string, RANGES> egressToRangesReq;
  for (auto& [egress, ranges] : egressToRanges_) {
    RANGES intersection;
    for (auto& range : ranges) {
      for (auto& r : reqToTuple_[reqName].first) {
        long lb = std::max(range.first, r.first);
        long ub = std::min(range.second, r.second);
        if (ub > lb) {
          intersection.push_back(std::make_pair(lb, ub));
        }
      }
    }
    egressToRangesReq[egress] = intersection;
  }
}

std::map<std::string, RANGES> MapReducer::map(folly::CIDRNetwork hs) {
  RANGE range = computeRange(hs);
  std::map<std::string, RANGES> egressToRangesReq;
  for (auto& [egress, ranges] : egressToRanges_) {
    RANGES intersection;
    for (auto& r : ranges) {
      long lb = std::max(range.first, r.first);
      long ub = std::min(range.second, r.second);
      if (ub > lb) {
        intersection.push_back(std::make_pair(lb, ub));
      }
    }
    egressToRangesReq[egress] = intersection;
  }
  return egressToRangesReq;
}

void MapReducer::reduce1() {
  // TODO
}

void MapReducer::reduce(RANGES input) {
  std::map<std::string, RANGES> nodeToHs;
  nodeToHs["A"] = {computeRange("11.0.0.0", 23)};
  nodeToHs["B"] = {computeRange("11.0.0.0", 24)};
  nodeToHs["C"] = {computeRange("11.0.1.0", 24)};
  nodeToHs["D"] = {computeRange("11.0.0.0", 23)};
  std::map<std::string, std::map<std::string, int>> meta;
  meta["A"]["eth0"] = 100;
  meta["A"]["eth1"] = 100;
  meta["B"]["eth0"] = 100;
  meta["C"]["eth0"] = 50;
  for (auto& [r, ar] : reqToAtomicRoutes_) {
    bool hasErr = false;
    for (auto& [egress, hs] : ar) {
      if (meta.find(nodeId_) == meta.end()) {
        break;
      }
      if (meta[nodeId_][egress] < 100 && intersection(nodeToHs[nodeId_], input).size() != 0) {
        hasErr = true;
        XLOG(ERR) << " [DDM] requirement " << r << " *NOT* satisfied at " + nodeId_;
      }
    }
    if (!hasErr && intersection(nodeToHs[nodeId_], input).size() != 0) {
      XLOG(INFO) << " [DDM] requirement " << r << " is satisfied at " + nodeId_;
    }
  }
}
}  // namespace ddm